package com.example.kafka.service;

import com.example.kafka.model.Order;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

/**
 * 订单消费者服务
 * 注意：此实现没有处理消息幂等性，可能会重复处理相同的消息
 */
@Service
public class OrderConsumerService {
    
    private static final Logger logger = LoggerFactory.getLogger(OrderConsumerService.class);
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    @KafkaListener(topics = "order-topic", groupId = "order-group")
    public void consumeOrder(String message, Acknowledgment ack) {
        try {
            // 解析订单消息
            Order order = objectMapper.readValue(message, Order.class);
            
            // 处理订单逻辑
            processOrder(order);
            
            // 手动确认消息
            ack.acknowledge();
            
            logger.info("订单处理成功: {}", order.getOrderId());
        } catch (Exception e) {
            // 处理失败，记录错误日志
            logger.error("订单处理失败", e);
            // 注意：这里不确认消息，消息会被重新投递
        }
    }
    
    /**
     * 处理订单业务逻辑
     * 注意：此方法可能会重复处理相同的订单
     */
    private void processOrder(Order order) {
        // 模拟订单处理
        logger.info("开始处理订单: {}", order.getOrderId());
        
        // TODO: 实现订单处理逻辑
        // 1. 更新订单状态
        // 2. 发送通知
        // 3. 其他业务逻辑
        
        // 模拟处理延迟
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("订单处理被中断", e);
        }
    }
} 